package com.lagou.boot;

import com.lagou.client.RPCConsumer;
import com.lagou.service.IUserService;
import com.lagou.service.UserServiceImpl;
import io.netty.channel.EventLoopGroup;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class ConsumerBoot {

    private static RPCConsumer rpcConsumer;

    public static List<String>[] zkListener(CuratorFramework client, Map<Long, IUserService> serviceMap, final List<String>[] childrenList) throws Exception {
        //注册对子节点的监听
        TreeCache cache = new TreeCache(client, "/netty");
        cache.start();
        TreeCacheListener tcl = (curatorFramework, treeCacheEvent) -> {
            //清空map
            serviceMap.clear();
            childrenList[0] = curatorFramework.getChildren().forPath("/netty");
            for (int i = 0; i < childrenList[0].size(); i++){
                int port = Integer.parseInt(childrenList[0].get(i));
                String host = new String(client.getData().forPath("/netty/" + childrenList[0].get(i)));
                Stat stat = new Stat();
                client.getData().storingStatIn(stat).forPath("/netty/" + childrenList[0].get(i) + "/time");
                long createTime = stat.getCtime();
                //1.创建代理对象
                rpcConsumer = new RPCConsumer();
                IUserService service = (IUserService) rpcConsumer.createProxy(IUserService.class, host, port);
                serviceMap.put(createTime, service);
            }
        };
        //注册监听
        cache.getListenable().addListener(tcl);
        return childrenList;
    }

    public static void main(String[] args) throws Exception {
        //连接服务端，从zookeeper获取连接信息
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(50000)
                .connectionTimeoutMs(30000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .build();
        client.start();
        List<String>[] childrenList = new List[]{client.getChildren().forPath("/netty")};
        Map<Long, IUserService> serviceMap = new HashMap<>();
        for (int i = 0; i < childrenList[0].size(); i++){
            int port = Integer.parseInt(childrenList[0].get(i));
            String host = new String(client.getData().forPath("/netty/" + childrenList[0].get(i)));
            Stat stat = new Stat();
            client.getData().storingStatIn(stat).forPath("/netty/" + childrenList[0].get(i) + "/time");
            long createTime = stat.getCtime();
            //1.创建代理对象
            rpcConsumer = new RPCConsumer();
            IUserService service = (IUserService) rpcConsumer.createProxy(IUserService.class, host, port);
            serviceMap.put(createTime, service);
        }

        childrenList = zkListener(client,serviceMap,childrenList);

        //2.循环给服务器写数据
        while(true){
            List<Long> createTimeList = new ArrayList<>(serviceMap.keySet());
            Long min = Collections.min(createTimeList);
            Long max = Collections.max(createTimeList);
            String result = serviceMap.get(min).sayHello("快给我success！！");
            SimpleDateFormat format =  new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSXXX"); //设置格式
            String minText=format.format(min);                                //获得带格式的字符串
            String maxText=format.format(max);                                //获得带格式的字符串
            System.out.println("节点创建时间为："+ minText + "-------" + result + "---------另一个节点创建时间为：" + maxText);
            Thread.sleep(2000);
        }

    }
}
